package com.hanson.demo.actors.system;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.function.Supplier;

import com.hanson.demo.actors.Actor;
import com.hanson.demo.actors.ActorMailbox;
import com.hanson.demo.actors.ActorRef;
import com.hanson.demo.actors.Dispatcher;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ActorSystem {

	/**
	 * 创建一个容器将 ActorRef 放到Map容器中
	 */
	private final ConcurrentMap<String, ActorRef> actorRefs = new ConcurrentHashMap<>();
	private final ConcurrentMap<String, Dispatcher> dispatchers = new ConcurrentHashMap<>();

	public void stop(ActorRef actorRef) {
		actorRef.stop();
		System.out.println("---关闭---");
	}

	public ActorRef createActor(String dispatcherId, Supplier<Actor> supplier) {
		Dispatcher dispatcher = dispatchers.get(dispatcherId);
		if (dispatcher == null) {
			log.warn("Dispatcher with id [{}] is not registered!", dispatcherId);
			// 初始化任务
			 createDispatcher(dispatcherId, initDispatcherExecutor(dispatcherId, 2));
			 dispatcher = dispatchers.get(dispatcherId);
		}
		Actor actor = this.createActor(supplier);
		ActorMailbox mailbox = new ActorMailbox(actor, dispatcherId, dispatcher);
		try {
			actor.initMail(mailbox); //actor 装配 mailbox
			mailbox.startActor(); //mailbox 初始化/启动等;
		} catch (Exception e) {
			e.printStackTrace();
		}
		ActorRef actorRef = actor.getActorRef();
		actorRefs.put(dispatcherId, actorRef);
		return actorRef;
	}

	private ExecutorService initDispatcherExecutor(String dispatcherName, int poolSize) {
		if (poolSize == 0) {
			int cores = Runtime.getRuntime().availableProcessors();
			poolSize = Math.max(1, cores / 2);
		}
		if (poolSize == 1) {
			return Executors.newSingleThreadExecutor();
		} else {
			return newWorkStealingPool(poolSize, dispatcherName);
		}
	}

	public Actor createActor(Supplier<Actor> supplier) {
		return supplier.get();
	}


	public static ExecutorService newWorkStealingPool(int parallelism, String namePrefix) {
		return new ForkJoinPool(parallelism, new MyForkJoinWorkerThreadFactory(namePrefix), null, true);
	}

	public void createDispatcher(String dispatcherId, ExecutorService executor) {
		Dispatcher current = dispatchers.putIfAbsent(dispatcherId, new Dispatcher(dispatcherId, executor));
		if (current != null) {
			throw new RuntimeException("Dispatcher with id [" + dispatcherId + "] is already registered!");
		}
	}

}
